package com.hisense.ovcloud.asop.service;

import com.alibaba.fastjson.JSON;
import com.hisense.ovcloud.asop.config.Constants;
import com.hisense.ovcloud.asop.entity.ch.AllPairApplog;
import com.hisense.ovcloud.asop.entity.ch.JulinkAnalysisLog;
import com.hisense.ovcloud.asop.entity.ch.K8sApplog;
import com.hisense.ovcloud.asop.entity.ch.WifiStatus;
import com.hisense.ovcloud.asop.tools.BeanReflectUtil;
import io.micrometer.common.util.StringUtils;
import io.r2dbc.spi.ConnectionFactory;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.codec.digest.DigestUtils;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Mono;

import java.nio.charset.Charset;
import java.nio.charset.StandardCharsets;
import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;

@Service
@Slf4j
public class ClickhouseQueryService {
    @Value("${queryCHTimeSpanSeconds:300}")
    private long queryCHTimeSpanSeconds;
    @Value("${queryCHLimit:20}")
    private long queryCHLimit;

    private final static DateTimeFormatter df = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

    final
    ConnectionFactory connectionFactory;

    public ClickhouseQueryService(ConnectionFactory connectionFactory) {
        this.connectionFactory = connectionFactory;
    }
    private String disposeSql(String userIds, String wifiId, String deviceId){
        StringBuffer sql = new StringBuffer().append(Constants.SqlTemplates
                .PAIR_LOG_GROUP_BY_PAIRID_PREFIX);

        StringBuilder sqlMidium = new StringBuilder();
        if(StringUtils.isNotEmpty(userIds)){
            sqlMidium.append(String.format(" user_id in (%s) ", userIds));
        }

        if(StringUtils.isNotEmpty(wifiId)){
            if(!sqlMidium.isEmpty()){
                sqlMidium.append(" and ");
            }
            sqlMidium.append(String.format(" wifiid = '%s' ", wifiId));
        }

        if(StringUtils.isNotEmpty(deviceId)){
            if(!sqlMidium.isEmpty()){
                sqlMidium.append(" and ");
            }
            sqlMidium.append(String.format(" deviceid = '%s' ", deviceId));
        }

        if(!sqlMidium.isEmpty()){
            sqlMidium.append(" and ");
        }else{
            sqlMidium.append(String.format(" wifiid = '%s' and ", Constants.WIFI_ID_CANNOT_BE_FOUND));
        }
        sqlMidium.append(Constants.SqlTemplates
                .PAIR_LOG_GROUP_BY_PAIRID_MEDIUM);

        sql.append(sqlMidium);
        sql.append(Constants.SqlTemplates
                .PAIR_LOG_GROUP_BY_PAIRID_SUFFIX);
        return sql.toString();
    }

    public Mono<List<AllPairApplog>> filterPairAppLogAndGroupByPairId(String type, String userIds, String beginTime, String endTime, String wifiId, String deviceId){
        return Mono.from(connectionFactory.create())
                .flatMapMany(conn -> {
                    String sqlTemp = disposeSql(userIds, wifiId, deviceId);
                    log.info(sqlTemp);
                    log.info(beginTime);
                    log.info(endTime);
                    return conn.createStatement(sqlTemp)
                            .bind("type", type)
                            .bind("beginTime", beginTime)
                            .bind("endTime", endTime)
                            .execute();
                })
                .flatMap(result -> {
                    if(result != null){
                        return result.map((row, rowMetadata) -> {
                            try {
                                return BeanReflectUtil.ckRowToBean(AllPairApplog.class, row);
                            } catch (Exception e) {
                                log.error("failed to map row to bean {}", "AllPairApplog");
                                throw new RuntimeException(e);
                            }
                        });
                    }else{
                        return Mono.just(AllPairApplog.builder().build());
                    }
                }).collectList();
    }

    public Mono<List<AllPairApplog>> selectByCdcTypeAndTime(String type, String mail, String beginTime, String endTime){
        return Mono.from(connectionFactory.create())
                .flatMapMany(conn -> {
                    String sqlTempl = Constants.SqlTemplates
                            .SELECT_BY_TYPE_AND_TIME
                            .replaceAll("__MAIL_MD5__", DigestUtils.md5Hex(mail));
                    log.info(sqlTempl);
                    log.info(beginTime);
                    log.info(endTime);
                    return conn.createStatement(sqlTempl)
                            .bind("type", type)
                            .bind("beginTime", beginTime)
                            .bind("endTime", endTime)
                            .execute();
                })
                .flatMap(result -> {
                    if(result != null){
                        return result.map((row, rowMetadata) -> {
                            try {
                                return BeanReflectUtil.ckRowToBean(AllPairApplog.class, row);
                            } catch (Exception e) {
                                log.error("failed to map row to bean {}", "AllPairApplog");
                                throw new RuntimeException(e);
                            }
                        });
                    }else{
                        return Mono.just(AllPairApplog.builder().build());
                    }
                }).collectList();
    }

    public Mono<List<AllPairApplog>> findPairAppLogByPairId(String pairid, String type, String beginTime, String endTime){
        return Mono.from(connectionFactory.create())
                .flatMapMany(conn -> {
                    String sql = Constants.SqlTemplates.FIND_PAIR_LOG_BY_PAIRID;
                    log.info(sql);
                    log.info(pairid);
                    log.info(type);
                    return conn.createStatement(sql)
                            .bind("pairid", pairid)
                            .bind("type", type)
                            .bind("beginTime", beginTime)
                            .bind("endTime", endTime)
                            .execute();
                })
                .flatMap(result -> {
                    if(result != null){
                        return result.map((row, rowMetadata) -> {
                            try {
                                return BeanReflectUtil.ckRowToBean(AllPairApplog.class, row);
                            } catch (Exception e) {
                                log.error("failed to map row to bean {}", "AllPairApplog");
                                throw new RuntimeException(e);
                            }
                        });
                    }else{
                        return Mono.just(AllPairApplog.builder().build());
                    }
                }).collectList();
    }

    public Mono<List<WifiStatus>> findWifistatusByWifiId(String wifiList, String beginTime, String endTime){
        return Mono.from(connectionFactory.create())
                .flatMapMany(conn -> {
                    String sql = Constants.SqlTemplates
                            .WIFI_STATUS_SEARCH
                            .replaceAll("__WIFI_IDS__", wifiList);
                    log.info(sql);
                    log.info(beginTime);
                    log.info(endTime);
                    return conn.createStatement(sql)
                            .bind("beginTime", beginTime)
                            .bind("endTime", endTime)
                            .execute();
                })
                .flatMap(result -> {
                    if(result != null){
                        return result.map((row, rowMetadata) -> {
                            try {
                                return BeanReflectUtil.ckRowToBean(WifiStatus.class, row);
                            } catch (Exception e) {
                                log.error("failed to map row to bean {}", "WifiStatus");
                                throw new RuntimeException(e);
                            }
                        });
                    }else{
                        return Mono.just(WifiStatus.builder().build());
                    }
                }).collectList();
    }

    public Mono<List<JulinkAnalysisLog>> queryJulinkLogs(String wifiId, long logTime){
        return Mono.from(connectionFactory.create())
                .flatMapMany(conn -> {
                    String beginTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(logTime - queryCHTimeSpanSeconds * 1000),
                            ZoneId.of("UTC")).atZone(ZoneId.of("UTC")).format(df);
                    String endTime = LocalDateTime.ofInstant(Instant.ofEpochMilli(logTime + queryCHTimeSpanSeconds * 1000),
                            ZoneId.of("UTC")).atZone(ZoneId.of("UTC")).format(df);;
;                   String sql = Constants.SqlTemplates
                            .QUERY_AC_JULINK_ANALYSIS_LOG
                            .replaceAll("__WIFIID__", wifiId)
                            .replaceAll("__BEGIN_TIME__", beginTime)
                            .replaceAll("__END_TIME__", endTime)
                            .replaceAll("__LIMIT__", String.valueOf(queryCHLimit));
                    log.info(sql);
                    log.info(beginTime);
                    log.info(endTime);
                    return conn.createStatement(sql)
                            .execute();
                })
                .flatMap(result -> {
                    if(result != null){
                        return result.map((row, rowMetadata) -> {
                            try {
                                return BeanReflectUtil.ckRowToBean(JulinkAnalysisLog.class, row);
                            } catch (Exception e) {
                                log.error("failed to map row to bean {}", "K8sApplog");
                                throw new RuntimeException(e);
                            }
                        });
                    }else{
                        return Mono.just(JulinkAnalysisLog.builder().build());
                    }
                }).collectList();
    }
}
